#include "pending_queue.h"
#include <zookeeper.h>
#include <proto.h>
#include <sys/types.h>

long get_time(){
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return tv.tv_sec*1000 + tv.tv_usec/1000;
}

pending_queue_t* pending_queue_init(){
    pending_queue_t * p_queue = NULL;
    p_queue = calloc(1, sizeof(*p_queue));
    p_queue->index = -1;
    p_queue->count = 0;
    p_queue->first_nonce = -1;
    pthread_mutex_init(&p_queue->lock,0);
    pthread_cond_init(&p_queue->cond,0);
    return p_queue;
}

void add_to_tail(pending_queue_t* p_queue, void *request, int nonce, int type, int timeout){
//   pthread_mutex_lock(&p_queue->lock);
    if(p_queue->first_nonce != -1 && nonce != p_queue->first_nonce + p_queue->count){
	printf("nonce expected %d, but got %d\n", p_queue->first_nonce + p_queue->count, nonce);
	exit(-1);
    } 
//    while(p_queue->count>=OUTSTANDING_LIMIT){
//	pthread_cond_wait(&p_queue->cond, &p_queue->lock);
//    }
    if(p_queue->first_nonce == -1)
    	p_queue->first_nonce = nonce;
    p_queue->count++;
    int pos = (p_queue->index + p_queue->count) % OUTSTANDING_LIMIT;
    //printf("add nonce %d to pos %d\n", nonce, pos);
    p_queue->requests[pos] = request;
    p_queue->request_type[pos] = type;
    p_queue->next_deadline[pos] = get_time() + timeout;

//    pthread_mutex_unlock(&p_queue->lock);

}

void* remove_first(pending_queue_t* p_queue){
//    pthread_mutex_lock(&p_queue->lock);
    if(p_queue->count <=0){
	printf("remove_first an empty queue\n");
	exit(-1);
    }
    p_queue->count --;
    p_queue->first_nonce ++;
   // void * ret = p_queue->requests[p_queue->index];
    p_queue->index = (p_queue->index + 1) % OUTSTANDING_LIMIT;
//    pthread_cond_broadcast(&p_queue->cond);
//    pthread_mutex_unlock(&p_queue->lock);
    return p_queue->requests[p_queue->index];
}

void* get_by_nonce(pending_queue_t* p_queue, int nonce){
//    pthread_mutex_lock(&p_queue->lock);
    void * ret = NULL;
    if(nonce < p_queue->first_nonce || nonce >= p_queue->first_nonce + p_queue->count){
	if(DEBUG_SHARD)
	printf("get_by_nonce out of range, nonce = %d, range = [%d, %d]\n",
		nonce, p_queue->first_nonce, p_queue->first_nonce + p_queue->count - 1);
//	pthread_mutex_unlock(&p_queue->lock);
	return NULL;
    }

    int pos = (nonce - p_queue->first_nonce + p_queue->index+1)%OUTSTANDING_LIMIT;
    ret = p_queue->requests[pos];
    //printf("get nonce %d from pos %d\n", nonce, pos);

//    pthread_mutex_unlock(&p_queue->lock);
    return ret;
}

void* get_by_index(pending_queue_t *p_queue, int index, int* type){
    void * ret = NULL;
    if(index < 0 || index > p_queue->count){
	printf("index %d out of bound %d\n", index, p_queue->count);
	return NULL;
    }
    int pos = (index + p_queue->index + 1)%OUTSTANDING_LIMIT;
    ret = p_queue->requests[pos];
    *type = p_queue->request_type[pos];
    return ret;
}


void check_timeout(block_storage_t *storage){
    int index;
    long now = get_time();
    pending_queue_t *p_queue = storage->pending_queue;
    if(DEBUG_SHARD)
    fprintf(stderr, "check timeout %d\n",p_queue->count);
    check_pending_queue(storage, 1);
    for(index = 0; index<p_queue->count; index++){
	int pos = (index + p_queue->index + 1)%OUTSTANDING_LIMIT;
	if(p_queue->next_deadline[pos] < now){
	    //printf("deadline is %d now is %d\n", p_queue->next_deadline[pos], now);
	    if(p_queue->request_type[pos]==PENDING_WRITE){
		if(pending_write_ready(p_queue->requests[pos], storage, 1)==0){
		    pending_write_timeout(storage, p_queue->requests[pos]);
		}
	    }
	    else {
		if(pending_read_ready(p_queue->requests[pos])==0)
		    pending_read_timeout(storage, p_queue->requests[pos]);
	    }
	    p_queue->next_deadline[pos] = now + zoo_recv_timeout(storage->servers);	
	}
    }
}

int is_empty(pending_queue_t* p_queue){

    return 1;
}

int is_full(pending_queue_t* p_queue){

    return 0;
}


/**
 * For pending write
 * */

pending_write_t* alloc_pending_write(long blockId, int xid, int nonce, char *buf, int buf_len,
     void_completion_t completion, void * data){
    pending_write_t * ret = NULL;
    ret = calloc(1, sizeof(*ret));
    if(ret==NULL){
	printf("alloc pending_write failed\n");
	exit(-1);
    }
    ret->blockId = blockId;
    ret->xid = xid;
    ret->nonce = nonce;
    ret->buf = buf;
    ret->buf_len = buf_len;
    ret->completion = completion;
    ret->data = data;
    int i;
    for(i=0; i<SERVER_COUNT; i++){
	ret->prepare_received[i]=0;
	ret->write_received[i]=0;
    }
    ret->write_sent = 0;
    pthread_mutex_init(&ret->lock,0);
    return ret;
}

void dealloc_pending_write(pending_write_t* p_write){
    pthread_mutex_destroy(&p_write->lock);
}

void pending_write_add_prepare_reply(pending_write_t* p_write, int serverId){
    pthread_mutex_lock(&p_write->lock);
    p_write->prepare_received[serverId] = 1;
    pthread_mutex_unlock(&p_write->lock);
}

int pending_write_data_ready(pending_write_t* p_write, block_storage_t *storage){
    int count = 0, i=0;
    int expected = get_num_live_servers(storage->servers);
    if(expected<2)
	expected = 2;
    if(expected>WRITE_QUORUM_SIZE)
        expected = WRITE_QUORUM_SIZE;
    pthread_mutex_lock(&p_write->lock);
    for(i=0; i<SERVER_COUNT; i++)
	count+=p_write->prepare_received[i];
    pthread_mutex_unlock(&p_write->lock);
    return count>=expected;
}

void pending_write_add_write_reply(pending_write_t* p_write, int serverId){
    pthread_mutex_lock(&p_write->lock);
    p_write->write_received[serverId] = 1;
    pthread_mutex_unlock(&p_write->lock);

}

int pending_write_ready(pending_write_t* p_write, block_storage_t *storage, int timeout){
    int count = 0, i=0;
    int expected = get_num_live_servers(storage->servers);
    if(expected<2)
        expected = 2;
    pthread_mutex_lock(&p_write->lock);
    for(i=0; i<SERVER_COUNT; i++)
        count+=p_write->write_received[i];
    pthread_mutex_unlock(&p_write->lock);
    if(timeout==0)
        return count>=expected;
    else
	return count>=2;
}


void pending_write_timeout(block_storage_t* storage, pending_write_t* p_write){
    pthread_mutex_lock(&p_write->lock);
    fprintf(stderr, "write nonce %d timeout %d\n", p_write->nonce, get_num_live_servers(storage->servers));
    /*if(p_write->write_sent){
	pthread_mutex_unlock(&p_write->lock);
	return;
    }*/
    int i;
    int nodes[SERVER_COUNT];
    int quorum[WRITE_QUORUM_SIZE];
    get_live_servers(storage->servers, nodes);
    int list = keyToNode(p_write->blockId, nodes, quorum);

    //for(i=0; i<WRITE_QUORUM_SIZE; i++){
        struct oarchive *oa;
        struct RequestHeader h = { .xid = p_write->xid, .nonce = p_write->nonce, .type = PREPAREDATA_OP};
        struct PrepareDataRequest req;
        int rc;

        oa = create_buffer_oarchive();
	req.clientId = storage->client_id;
	req.count = list;//WRITE_QUORUM_SIZE;
	fprintf(stderr, "resend prepare to %d\n", list);
        req.data.buff = p_write->buf;
        req.data.len = p_write->buf_len;
        rc = serialize_RequestHeader(oa, "header", &h);
        rc = rc < 0 ? rc : serialize_PrepareDataRequest(oa, "req", &req);

        send_bytes_to_server(storage->servers, get_buffer(oa),
            get_buffer_len(oa), quorum[0]);
        if(DEBUG_SHARD)
        printf("sending prepare nonce =%#x for blockId [%d] to %d\n",h.nonce,p_write->blockId, quorum[0]);
        close_buffer_oarchive(&oa, 0);
    //}
    pthread_mutex_unlock(&p_write->lock);

}

/**
 * For pending read
 * */

pending_read_t* alloc_pending_read(long blockId, int nonce, data_completion_t completion, void *data){
    pending_read_t* ret = NULL;
    ret = calloc(1, sizeof(*ret));
    ret->blockId = blockId;
    ret->nonce = nonce;
    ret->completion = completion;
    ret->data = data;
    int i;
    for(i=0; i<SERVER_COUNT; i++)
	ret->node_received[i] = 0;
    pthread_mutex_init(&ret->lock,0);
    return ret;
}

void dealloc_pending_read(pending_read_t* p_read){
    pthread_mutex_destroy(&p_read->lock);
}

void pending_read_add_reply(pending_read_t* p_read, int serverId){
    pthread_mutex_lock(&p_read->lock);
    p_read->node_received[serverId]=1;
   //printf("add read reply nonce %d\n", p_read->nonce);
    pthread_mutex_unlock(&p_read->lock);

}

int pending_read_ready(pending_read_t* p_read){
    int count = 0, i=0;
    pthread_mutex_lock(&p_read->lock);
    for(i=0; i<SERVER_COUNT; i++)
        count+=p_read->node_received[i];
    pthread_mutex_unlock(&p_read->lock);
    //printf("read count nonce %d = %d\n", p_read->nonce, count);
    return count>=1;
}

void pending_read_timeout(block_storage_t* storage, pending_read_t* p_read){
    pthread_mutex_lock(&p_read->lock);
    printf("read nonce %d timeout\n", p_read->nonce);
    struct oarchive *oa;
    struct RequestHeader h = { .xid = -1, .nonce = p_read->nonce, .type = BOLOSKYGETDATA_OP};
    struct BoloskyGetDataRequest req;
    int rc;
    oa = create_buffer_oarchive();
    req.blockId = p_read->blockId;
    int nodes[SERVER_COUNT];
    get_live_servers(storage->servers, nodes);
    int i=0;
    int index = -1;
    for(i=1; i<SERVER_COUNT; i++){
	index = (p_read->serverId-i+SERVER_COUNT)%SERVER_COUNT;
	if(nodes[index]==1)
	    break;
    }
    if(index == -1){
	printf("read connot find live servers\n");
	exit(-1);
    }
    req.serverId = index;
    p_read->serverId = index;
    rc = serialize_RequestHeader(oa, "header", &h);
    rc = rc < 0 ? rc : serialize_BoloskyGetDataRequest(oa, "req", &req);

    send_bytes_to_server(storage->servers, get_buffer(oa),
            get_buffer_len(oa), SERVER_COUNT-1);
    if(DEBUG_SHARD)
    printf("sending read nonce=%#x for blockId [%d] to %d\n",h.nonce, p_read->blockId, SERVER_COUNT-1);
    close_buffer_oarchive(&oa, 0);
    pthread_mutex_unlock(&p_read->lock);
}

